#pragma once
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include <functional>
#include <iostream>
#include <thread>

#include "logger.hpp"

class MQClient
{
public:
    using ptr = std::shared_ptr<MQClient>;
    using MessageCallback = std::function<void(const char *, size_t)>;
    MQClient(const std::string &user,
             const std::string &passwd,
             const std::string host)
    {
        _loop = EV_DEFAULT;
        _handler = std::make_unique<AMQP::LibEvHandler>(_loop);
        std::string url = "amqp://" + user + ":" + passwd + "@" + host + "/";
        AMQP::Address address(url);
        // 创建连接对象指针
        _connection = std::make_unique<AMQP::TcpConnection>(_handler.get(), address);
        // 创建通道
        _channel = std::make_unique<AMQP::TcpChannel>(_connection.get());

        _loop_thread = std::thread([this]()
                                   { ev_run(_loop, 0); });
    }

    ~MQClient()
    {
        ev_async_init(&_ev_async_watcher, watcher_callback);
        ev_async_start(_loop, &_ev_async_watcher);
        ev_async_send(_loop, &_ev_async_watcher);
        _loop_thread.join();
        _loop = nullptr;
    }

    // 声明组件
    void declareComponents(const std::string &exchange_name,
                           const std::string &queue,
                           const std::string &routing_key = "routing_key",
                           AMQP::ExchangeType exchange_type = AMQP::ExchangeType::direct)
    {
        _channel->declareExchange(exchange_name, exchange_type)
            .onError([](const char *message)
                     {
                        LOG_ERROR("声明交换机失败");
        return -1; })
            .onSuccess([]()
                       { LOG_ERROR("声明交换机成功"); });

        // 声明队列
        _channel->declareQueue(queue)
            .onError([](const char *message)
                     {
                        LOG_ERROR("声明队列失败");
        return -1; })
            .onSuccess([]()
                       { LOG_INFO("声明队列成功"); });

        _channel->bindQueue(exchange_name, queue, routing_key)
            .onError([exchange_name, queue](const char *message){
                 LOG_ERROR("{}-{}绑定失败",exchange_name,queue);
                 exit(0);
                 })
            .onSuccess([exchange_name, queue, routing_key](){ 
                LOG_INFO("{} - {} - {} 绑定成功！", exchange_name, queue, routing_key); 
                });
    }

    bool publish(const std::string &exchange_name,
                 const std::string &msg,
                 const std::string &routing_key = "routing_key")
    {
        bool ret = _channel->publish(exchange_name, routing_key, msg);
        if (ret == false)
        {
            LOG_ERROR("{} 发布消息失败", exchange_name);
            return false;
        }

        return true;
    }

    bool consume(const std::string &queue, const MessageCallback &cb)
    {
        _channel->consume(queue, "consume-tag")
            .onReceived([this, cb](const AMQP::Message &message,
                                   uint64_t deliveryTag,
                                   bool redelivered)
                        {
        cb(message.body(),message.bodySize());

        _channel->ack(deliveryTag); })
            .onError([](const char *message)
                     {
        std::cout << "订阅test-queue队列 消息失败" << message << std::endl;
        return -1; });

        return true;
    }

private:
    static void watcher_callback(struct ev_loop *loop, ev_async *watcher, int32_t revents)
    {
        ev_break(loop, EVBREAK_ALL);
    }

private:
    struct ev_async _ev_async_watcher;
    struct ev_loop *_loop;
    std::unique_ptr<AMQP::LibEvHandler> _handler;
    std::unique_ptr<AMQP::TcpConnection> _connection;
    std::unique_ptr<AMQP::TcpChannel> _channel;
    std::thread _loop_thread;
};
